package com.huishi.handler.thread.task;

import com.huishi.entity.DataSource;
import com.huishi.util.CMDUtils;
import com.huishi.util.DBUtils;
import com.huishi.util.DataXUtils;
import com.huishi.util.JSONUtils;
import lombok.extern.slf4j.Slf4j;

import java.util.Date;
import java.util.Map;

/**
 * DataX增量同步任务
 */
@Slf4j
public class DataXIncrementalSyncTask implements Runnable {

    private DataSource source;
    private DataSource target;
    private String tableName;

    public DataXIncrementalSyncTask(DataSource source, DataSource target, String tableName) {
        this.source = source;
        this.target = target;
        this.tableName = tableName;
    }

    @Override
    public void run() {
        log.info("{}--开始执行DataX增量同步任务", tableName);
        Date startTime = new Date();

        // 0增量source -> target：获取target最大时间，source过滤大于这个时间的
        String sql = "select max(update_time) from " + tableName;
        Map<String, Object> result = DBUtils.executeSQLByCmd(target.getIp(), target.getPort(), target.getUserName(), target.getPassword(),
                target.getSchema(), sql);

        String maxUpdateTime = null;
        if ((boolean) result.get("result")) {
            maxUpdateTime = result.get("value").toString().split("\\)")[1];
            // 1生成json
            String jobName = JSONUtils.generateDataXJob(source, target, tableName, maxUpdateTime);

            // 2执行
            log.info("{}--DataX增量同步任务执行结果：{}", tableName, DataXUtils.executeJob(jobName, startTime));
        } else
            log.error("{}--DataX增量同步任务执行失败，查询最大时间戳失败。", tableName);
        System.out.println();
    }
}
